package cn.yma.bus;

import cn.yma.bus.annotation.MessageBus;
import cn.yma.bus.annotation.MessageSubscribe;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.function.Consumer;

@Slf4j
public class RedisBus implements ApplicationContextAware {
    Map<String, List<Consumer<String>>> methods = new HashMap<>();

    private final RedisMessageListenerContainer container;

    public RedisBus(RedisMessageListenerContainer container) {
        this.container = container;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        applicationContext
                .getBeansWithAnnotation(MessageBus.class)
                .values().forEach(bean -> {
                    for (Method method : bean.getClass().getMethods()) {
                        MessageSubscribe[] annotationsByType = method.getAnnotationsByType(MessageSubscribe.class);
                        for (MessageSubscribe messageSubscribe : annotationsByType) {
                            if (messageSubscribe.topic().length() > 0) {
                                addTopic(bean, method, messageSubscribe.topic());
                            }

                            if(messageSubscribe.topics().length > 0){
                                for (String topic : messageSubscribe.topics()) {
                                    addTopic(bean,method,topic);
                                }
                            }
                        }
                    }
                });

        methods.forEach((topic,list) ->
                container.addMessageListener((message, pattern) ->
                                list.forEach(consumer ->
                                        consumer.accept(message.toString())),new ChannelTopic(topic)));
    }

    /**
     * 订阅通道/主题
     * @param topic 通道/主题
     * @param consumer 消息处理器
     */
    public void addMessageListener(String topic,Consumer<String> consumer){
        container.addMessageListener((message, pattern) -> consumer.accept(message.toString()),new ChannelTopic(topic));
    }

    private void addTopic(Object bean, Method method, String topic) {
        List<Consumer<String>> consumers = methods.computeIfAbsent(topic, k -> new ArrayList<>());
        consumers.add(msg -> {
            try {
                method.invoke(bean, msg);
            } catch (IllegalAccessException e) {
                log.warn("访问权异常",e);
            } catch (InvocationTargetException e) {
                log.warn("内部异常",e);
            } catch (Exception e){
                log.warn("其它异常",e);
            }
        });
    }
}
